package org.gol.paxos.member;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.gol.paxos.Message;
import org.gol.paxos.Proposal;
import org.gol.paxos.Topic;
import org.gol.paxos.client.PaxosIslandBasic;
import org.gol.paxos.role.Learner;
import org.gol.paxos.role.MessageTranser;
import org.gol.paxos.role.Proposer;

/**
 * @package org.gol.paxos.client.member
 * @description 【公民代表】接收议题，在议会上向参议员提出提案
 * @author Goliath
 * @createTime 2012-11-21 下午1:38:31
 * @modifyTime
 */
public class Citizen implements Proposer, Learner {
	private String positionNo;
	private long proposalIndex = 0;
	private boolean standby = false;
	private ExecutorService exec = Executors.newCachedThreadPool();
	private BlockingQueue<Message> messageQueue = new ArrayBlockingQueue<Message>(10);
	/**
	 * 公民代表都带了个文件夹，来记录提案 【内容和状态】
	 */
	public Map<String, Proposal> folder = Collections.synchronizedMap(new HashMap<String, Proposal>());

	private Runnable getTopic = new Runnable() {
		public void run() {
			while (true) {
				// 每隔一段时间， 公民代表去公告栏看看有没有新的议题发布，并按新的议题整理出提案放到自己的文件夹里
				try {
					getNewTopic();
					Thread.sleep((int) (1000 * Math.random()));
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	};

	private Runnable checkFold = new Runnable() {
		public void run() {
			while (true) {
				try {
					// 不定期的查看一下自己的文件夹中的提案状态
					checkFold();
					Thread.sleep((int) (1000 * Math.random()));
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	};
	
	private Runnable receiveMessage = new Runnable() {
		public void run() {
			while (true) {
				try {
					Message oMessage = getMessageQueue().take();
					receive(oMessage);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	};

	public Citizen(String positionNo) {
		this.setPositionNo(positionNo);
	}

	private void checkFold() {
		List<Proposal> list = new ArrayList<Proposal>(this.folder.values());
		for (int i = 0; i < list.size(); i++) {
			Proposal oProposal = list.get(i);
			// 发现有一个需要处理的提案
			if (oProposal.getState() == Proposal.INIT) {

				// 给这个提案附上自己的值
				oProposal.setResult(randomResult());
				// 给这个提案搞一个序号
				oProposal.setProNo(this.getPositionNo() + "_P" + proposalIndex);
				proposalIndex++;
				sendPrepare(oProposal);
				oProposal.setState(Proposal.SENDED);
				return;
			}
		}
	}

	private void sendPrepare(Proposal oProposal) {
		// 先获取一个足够形成多数派的议员列表
		int majorityNum = getMajorityNum(PaxosIslandBasic.oParliament.getMaxSenatorNum());
		String[] sendArr = getSendList(majorityNum);
		// 复印列表size份数的提案副本
		// 找来列表size数量的侍从
		String str = this.positionNo + "整理了一个新建的提案" + oProposal + "提交参议员";
		for (String sendTo : sendArr) {
			str += sendTo + ",";
		}
		str += "批示";
		System.out.println(str);
		// 发送提案
		for (int i = 0; i < majorityNum; i++) {
			Message oMessage = new Message();
			oMessage.setCommand("prepare");
			oMessage.setFrom(this.getPositionNo());
			oMessage.setSendTo(sendArr[i]);
			try {
				// 复印一份提案
				oMessage.setContent(oProposal.clone());
			} catch (CloneNotSupportedException e) {
				e.printStackTrace();
				return;
			}
			MessageTranser oMessageTranser = new Servent();
			oMessageTranser.transMessage(oMessage);
		}
	}

	private String[] getSendList(int majorityNum) {
		Map<String, String> map = new HashMap<String, String>();
		String[] sendArrAll = PaxosIslandBasic.oParliament.acceptorPositions.keySet().toArray(new String[] {});
		while (map.size() < majorityNum) {
			int index = (int) (Math.random() * sendArrAll.length);
			if (!map.containsKey(sendArrAll[index])) {
				map.put(sendArrAll[index], null);
			}
		}
		return map.keySet().toArray(new String[] {});
	}

	private int getMajorityNum(int majorityMaxNum) {
		int majorityMinNum = majorityMaxNum / 2 + 1;
		return (int) (Math.random() * (1 + majorityMaxNum - majorityMinNum)) + majorityMinNum;
	}

	private int randomResult() {
		int r = (int) (Math.random() * 3) + 1;
		if (r == 1) {
			return 100;
		} else if (r == 2) {
			return 200;
		} else if (r == 3) {
			return 500;
		}
		return 0;
	}

	public String getPositionNo() {
		return positionNo;
	}

	public void setPositionNo(String positionNo) {
		this.positionNo = positionNo;
	}

	/**
	 * @package org.gol.paxos.member
	 * @description 获取新议题，并生成提案
	 * @returnType void
	 * @author Goliath
	 * @createTime 2012-11-22 上午9:42:01
	 * @modifyTime
	 */
	private void getNewTopic() {
		for (int i = 0; i < PaxosIslandBasic.topicList.size(); i++) {
			Topic oTopic = PaxosIslandBasic.topicList.get(i);
			// 当自己的文件夹中没有这个议题
			if (!this.folder.containsKey(oTopic.getContent())) {
				createProposal(oTopic);
			}
		}
	}

	/* (non-Javadoc)
	 * @see org.gol.paxos.role.Proposer#createProposal(org.gol.paxos.Topic)
	 */
	@Override
	public void createProposal(Topic oTopic) {
		// System.out.println(this.getPositionNo() + "创建了关于[" +
		// oTopic.getContent() + "]的提案");

		Proposal oProposal = new Proposal(oTopic.getContent());
		this.folder.put(oProposal.getContent(), oProposal);
	}

	@Override
	public void standBy() {
		if (this.standby == false) {
			exec.submit(getTopic);
			exec.submit(checkFold);
			exec.submit(receiveMessage);
			this.standby = true;
		}
	}

	@Override
	public void receive(Message oMessage) {
		if ("agree".equals(oMessage.getCommand())) {
			Proposal oProposalCopy = (Proposal) oMessage.getContent();
			Proposal oProposal = this.folder.get(oProposalCopy.getContent());
			System.out.println(oProposal + " agree by " + oMessage.getFrom());
			if (oProposal.getProNo().equals(oProposalCopy.getProNo())) {
				oProposal.getAgreeCopy().add(oProposalCopy);
				if (oProposal.getState() != Proposal.CONFIRM) {
					if (oProposal.getAgreeCopy().size() >= PaxosIslandBasic.oParliament.getMaxSenatorNum() / 2 + 1) {
						oProposal.setState(Proposal.CONFIRM);
						System.out.print(oProposal + "已经由多数派:");
						for (int i = 0; i < oProposal.getAgreeCopy().size(); i++) {
							System.out.print(oProposal.getAgreeCopy().get(i).getAcceptorName() + ",");
						}
						System.out.println("通过了! " + oProposal + " ---- 进入Confirm阶段");
						sendConfirm(oProposal);
					}
				} else {
					System.out.println(oProposal + "由" + oMessage.getFrom() + "通过，但是已经在Confirm阶段");
				}
			}
		} else if ("disagree".equals(oMessage.getCommand())) {
			Proposal oProposalCopy = (Proposal) oMessage.getContent();
			Proposal oProposal = this.folder.get(oProposalCopy.getContent());
			System.out.println(oProposal + " disagree by " + oMessage.getFrom() + ",reason:"
					+ oProposalCopy.getComment());
			if (oProposal.getProNo().equals(oProposalCopy.getProNo())) {
				oProposal.getDisagreeCopy().add(oProposalCopy);
				if (oProposal.getDisagreeCopy().size() >= PaxosIslandBasic.oParliament.getMaxSenatorNum() / 2 + 1
						&& oProposal.getState() != Proposal.DISAGREE) {
					oProposal.setState(Proposal.DISAGREE);
					System.out.print(oProposal + "被多数派:");
					for (int i = 0; i < oProposal.getDisagreeCopy().size(); i++) {
						System.out.print(oProposal.getDisagreeCopy().get(i).getAcceptorName() + ",");
					}
					System.out.println("驳回了!");
				}
			}
		} else {
			System.out.println("do nothing");
		}
	}

	private void sendConfirm(Proposal oProposal) {
		String[] sendArr = PaxosIslandBasic.oParliament.acceptorPositions.keySet().toArray(new String[] {});
		// 不考虑优化算法，给所有人都发一份
		for (int i = 0; i < sendArr.length; i++) {
			Message oMessage = new Message();
			oMessage.setCommand("confirm");
			oMessage.setFrom(this.getPositionNo());
			oMessage.setSendTo(sendArr[i]);
			oMessage.setContent(oProposal);
			MessageTranser oMessageTranser = new Servent();
			oMessageTranser.transMessage(oMessage);
			System.out.println(this.getPositionNo() + "提交confirm " + oProposal + "至" + sendArr[i]);
		}
	}

	public BlockingQueue<Message> getMessageQueue() {
		return messageQueue;
	}

	public void setMessageQueue(BlockingQueue<Message> messageQueue) {
		this.messageQueue = messageQueue;
	}
}
